package com.yinanxi.lflume;

import com.google.gson.Gson;
import org.apache.commons.lang.time.FastDateFormat;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;

/**
 * @Created by Orion
 * @Description
 *  自定义拦截器
 *    1) 抽取日志数据中的时间戳. 将时间戳转换成日期  timestamp
 *    2) 将日期 设置在Header中
 */
public class ExtractLogDateInterceptor implements Interceptor {
    String  key = null ;
    Gson gson = null ;
    FastDateFormat format = null ;
    public ExtractLogDateInterceptor(String  key) {
        this.key = key ;
    }
    /**
     * 拦截器初始化的方法  只执行一次
     */
    public void initialize() {
        gson = new Gson();
        format = FastDateFormat.getInstance("yyyy-MM-dd");
    }

    /**
     * @param event   拦截一个Event处理
     * @return
     */
    public Event intercept(Event event) {
        try {
            // 脱敏 密码  邮箱 敏感数据  加密
            byte[] body = event.getBody();
            // 获取日志行数据
            String jsonRow = new String(body);
            // 从行数据中获取 timestamp
            LogBean logBean = gson.fromJson(jsonRow, LogBean.class);
            Long timestamp = logBean.getTimestamp();  //1652154631125
            // 将时间戳转换成日期
            String date = this.format.format(timestamp);
            //将日期格式的字符串  添加到Header中
            event.getHeaders().put(key, date) ;  //%{date}
        } catch (Exception e) {
            return null ;
        }
        return event;
    }

    /**
     *
     * @param events  拦截所有的事件
     * @return
     */
    public List<Event> intercept(List<Event> events) {
        // 为没一个事件添加当前日志的日期
        for (Event event : events) {
            intercept(event) ;
        }
        return events;
    }
    /**
     * 结束的方法 只执行一次
     * 收尾工作 比如释放连接  释放对象
     */
    public void close() {
    gson = null ;
    format = null ;
    }
    /**
     * 拦截器构建器
     *  必须是 public   static
     *  com.yinanxiedu.lflume.ExtractLogDateInterceptor$ExtractLogDateInterceptorBuilder
     *
     */
    public  static  class ExtractLogDateInterceptorBuilder implements  Interceptor.Builder{
        String key = null ;
        public Interceptor build() {
            return new ExtractLogDateInterceptor(key);
        }
       // 获取值    配置信息  a1.sources.s1.interceptors.i1.key  = date
        public void configure(Context context) {
             key = context.getString("key");
        }
    }
}
